package com.dayang.redbook.count.biz.consumer;

import cn.hutool.core.util.StrUtil;
import com.dayang.framework.common.util.JsonUtil;
import com.dayang.redbook.count.biz.constant.MQConstants;
import com.dayang.redbook.count.biz.domain.mapper.UserCountDOMapper;
import com.dayang.redbook.count.biz.enums.FollowUnfollowTypeEnum;
import com.dayang.redbook.count.biz.model.dto.CountFollowUnfollowMqDTO;
import com.google.common.util.concurrent.RateLimiter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * @author: dayang
 * @since: 2024/12/24
 * @description: 计数: 关注数存库
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_COUNT_FOLLOWING_2_DB,
        topic = MQConstants.TOPIC_COUNT_FOLLOWING_2_DB
)
@Slf4j
@RequiredArgsConstructor
public class CountFollowing2DBConsumer implements RocketMQListener<String> {
    private final UserCountDOMapper userCountDOMapper;
    private RateLimiter rateLimiter = RateLimiter.create(5000);
    @Override
    public void onMessage(String body) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();

        log.info("## 消费到了 MQ 【计数: 关注数入库】, {}...", body);

        if (StrUtil.isBlank(body)) return;

        CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = JsonUtil.parseObject(body, CountFollowUnfollowMqDTO.class);

        // 操作类型：关注 or 取关
        Integer type = countFollowUnfollowMqDTO.getType();
        // 原用户ID
        Long userId = countFollowUnfollowMqDTO.getUserId();

        // 关注数：关注 +1， 取关 -1
        int count = Objects.equals(type, FollowUnfollowTypeEnum.FOLLOW.getCode()) ? 1 : -1;
        // 判断数据库中，若原用户的记录不存在，则插入；若记录已存在，则直接更新
        userCountDOMapper.insertOrUpdateFollowingTotalByUserId(count, userId);
    }


}
